package c7;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.function.Function;
import java.util.stream.LongStream;
import java.util.stream.Stream;

public class ParallelStreams {
    public static void main(String[] args){
        System.out.println("iterativeSum done in "+measureSumPerf(ParallelStreams::iterativeSum,10_000_000));
        System.out.println("sequentialSum done in "+measureSumPerf(ParallelStreams::sequentialSum,10_000_000));
        System.out.println("parallelSum done in "+measureSumPerf(ParallelStreams::parallelSum,10_000_000));
        System.out.println("rangedSum done in "+measureSumPerf(ParallelStreams::rangedSum,10_000_000));
        System.out.println("rangedParallelSum done in "+measureSumPerf(ParallelStreams::rangedParallelSum,10_000_000));

        System.out.println("sideEffectSum done in "+measureSumPerf(ParallelStreams::sideEffectSum,10_000_000));
        System.out.println("sideEffectParallelSum done in "+measureSumPerf(ParallelStreams::sideEffectParallelSum,10_000_000));

        System.out.println("forkJoinSum done in "+measureSumPerf(ParallelStreams::forkJoinSum,10_000_000));
    }
    public static long measureSumPerf(Function<Long, Long> adder, long n) {
        long fastest = Long.MAX_VALUE;
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            long sum = adder.apply(n);
            long duration = (System.nanoTime() - start) / 1_000_000;
            System.out.println("Result: " + sum);
            if (duration < fastest) fastest = duration;
        }
        return fastest;
    }
    public static long iterativeSum(long n) {
        long result = 0;
        for (long i = 0; i <= n; i++) {
            result += i;
        }
        return result;
    }
    public static long sequentialSum(long n) {
        return Stream.iterate(1L, i -> i + 1).limit(n).reduce(Long::sum).get();
    }

    public static long parallelSum(long n) {
        return Stream.iterate(1L, i -> i + 1).limit(n).parallel().reduce(Long::sum).get();
    }

    public static long rangedSum(long n){
        return LongStream.rangeClosed(1,n)
                .reduce(0L,Long::sum);
    }

    public static long rangedParallelSum(long n){
        return LongStream.rangeClosed(1,n)
                .parallel()
                .reduce(0L,Long::sum);
    }
    public static long sideEffectSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).forEach(accumulator::add);
        return accumulator.total;
    }
    public static long sideEffectParallelSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);
        return accumulator.total;
    }
    public static class Accumulator {
        public long total = 0;
        public void add(long value) { total += value; }
    }

    public static long forkJoinSum(long n){
        long[] numbers=LongStream.rangeClosed(1,n).toArray();
        ForkJoinTask<Long> forkJoinTask=new ForkJoinSumCalculator(numbers);
        return new ForkJoinPool().invoke(forkJoinTask);
    }
}
